package logger

import (
	"encoding/json"

	kafka "github.com/Shopify/sarama"
)

//Type 日志类型
type Config struct {
	//broker连接地址
	Addrs []string `json:"addrs"`
	//主题
	//Topic string `json:"topic"`
	//分区
	Partition int32 `json:"partition"`
	//批量提交最大字节数
	MaxBytes int32 `json:"maxBytes"`
	//批量提交的最大消息个数
	MaxMsgCnt int `json:"maxMsgCnt"`
	//其他待补充
}

const brokerNamePrefix = "wind.platform.topic."

type Logger struct {
	producer  kafka.AsyncProducer
	topic     string
	partition int32
}

func NewLogger(typ string, c *Config) *Logger {
	var l Logger
	kc := kafka.NewConfig()
	kc.Producer.Return.Successes = false
	kc.Producer.Return.Errors = false
	kc.Consumer.Return.Errors = false
	//kc.Producer.Retry.Backoff = 0
	if c.MaxBytes != 0 {
		kafka.MaxRequestSize = c.MaxBytes
	}
	if c.MaxMsgCnt != 0 {
		kc.Producer.Flush.MaxMessages = c.MaxMsgCnt
	}

	var err error
	l.producer, err = kafka.NewAsyncProducer(c.Addrs, kc)
	if err != nil {
		//wplog.Error("init kafka faield:", err.Error())
		return nil
	}

	l.partition = c.Partition
	l.topic = brokerNamePrefix + typ

	return &l
}

//写日志
func (l *Logger) Log(d interface{}) {
	l.producer.Input() <- &kafka.ProducerMessage{
		Topic:     l.topic,
		Partition: l.partition,
		Key:       nil,
		Value:     newJsonEncoder(d),
	}
}

type jsonEncoder struct {
	b []byte
	e error
}

func newJsonEncoder(i interface{}) *jsonEncoder {
	r := &jsonEncoder{}
	r.b, r.e = json.Marshal(i)
	return r
}

func (j *jsonEncoder) Encode() ([]byte, error) {
	return j.b, j.e
}

func (j *jsonEncoder) Length() int {
	return len(j.b)
}
